package com.atguigu.flink.day09;

import com.atguigu.flink.bean.WaterSensor;
import com.atguigu.flink.func.WaterSensorMapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.runtime.state.storage.JobManagerCheckpointStorage;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author Felix
 * @date 2024/8/20
 * 该案例演示了算子状态-广播状态
 * 需求：水位超过指定的阈值发送告警，阈值可以动态修改
 */
public class Flink08_OpeState_BroadcastState {
    public static void main(String[] args) throws Exception {
        //TODO 1.环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(2);

        //env.setStateBackend(new HashMapStateBackend());

        //env.setStateBackend(new EmbeddedRocksDBStateBackend());

        //TODO 2.从指定的网络端口读取水位信息
        SingleOutputStreamOperator<WaterSensor> wsDS = env
                .socketTextStream("hadoop102", 8888)
                .map(new WaterSensorMapFunction());
        //TODO 3.从指定的网络端口读取阈值信息
        SingleOutputStreamOperator<Integer> thresholdDS = env
                .socketTextStream("hadoop102", 8889)
                .map(Integer::valueOf);
        //TODO 4.广播阈值流
        //注意：广播状态的使用方式是固定的，在对流中数据进行广播的时候，需要在Broadcast方法中传递状态描述器，描述状态中存储的数据结构
        MapStateDescriptor<String, Integer> mapStateDescriptor
                = new MapStateDescriptor<String, Integer>("mapStateDescriptor",String.class, Integer.class);
        BroadcastStream<Integer> broadcastDS = thresholdDS.broadcast(mapStateDescriptor);

        //TODO 5.将主流(水位信息)和广播流(阈值信息)进行关联---connect
        BroadcastConnectedStream<WaterSensor, Integer> connectDS = wsDS.connect(broadcastDS);

        //TODO 6.对连接后的数据进行处理---process
        //processElement:处理主流数据              从状态中获取阈值信息判断是否超过了警戒线
        //processBroadcastElement:处理广播流数据   将广播流中的阈值放到状态中存储起来
        SingleOutputStreamOperator<String> processDS = connectDS.process(
                new BroadcastProcessFunction<WaterSensor, Integer, String>() {
                    @Override
                    public void processElement(WaterSensor ws, BroadcastProcessFunction<WaterSensor, Integer, String>.ReadOnlyContext ctx, Collector<String> out) throws Exception {
                        //获取当前采集的水位信息
                        Integer vc = ws.getVc();
                        //获取广播状态  在处理主流数据的时候，只能使用广播状态，不能对其进行修改
                        ReadOnlyBroadcastState<String, Integer> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
                        //从广播状态中获取阈值信息
                        Integer threshold = broadcastState.get("threshold") == null ? 10000 : broadcastState.get("threshold");

                        if (vc > threshold) {
                            out.collect("当前水位值" + vc + "已经超过警戒线" + threshold + "赶快撤离！~");
                        }

                    }

                    @Override
                    public void processBroadcastElement(Integer threshold, BroadcastProcessFunction<WaterSensor, Integer, String>.Context ctx, Collector<String> out) throws Exception {
                        //获取广播状态
                        BroadcastState<String, Integer> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
                        //将阈值信息放到广播状态中
                        broadcastState.put("threshold", threshold);
                    }
                }
        );
        //TODO 7.打印输出
        processDS.print();
        //TODO 8.提交作业
        env.execute();
    }
}
